package com.etc.java;

import org.apache.commons.collections.IteratorUtils;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.api.java.function.VoidFunction;
import scala.Tuple2;


import java.util.Collections;
import java.util.Iterator;
import java.util.List;

/**
 * @author 杰哥
 * @Title: topN
 * @ProjectName scalaworldcount
 * @Description: TODO
 * @date 2019/7/239:41
 */
public class topN {
    public static void main(String[] args) {
        SparkConf conf = new SparkConf().setAppName("topN").setMaster("local");
        JavaSparkContext jsc = new JavaSparkContext(conf);
        JavaRDD<String> stringJavaRDD = jsc.textFile("E:\\input\\score.txt");
        JavaPairRDD<String, Integer> pairRDD = stringJavaRDD.mapToPair(new PairFunction<String, String, Integer>() {
            @Override
            public Tuple2<String, Integer> call(String s) throws Exception {
                String[] s1 = s.split(" ");
                return new Tuple2<String, Integer>(s1[0], Integer.valueOf(s1[1]));
            }
        });
        JavaPairRDD<String, Iterable<Integer>> jp = pairRDD.groupByKey();

        jp.foreach(new VoidFunction<Tuple2<String, Iterable<Integer>>>() {
            @Override
            public void call(Tuple2<String, Iterable<Integer>> sit) throws Exception {
                System.out.println(sit._1);
                Iterator<Integer> iterator = sit._2.iterator();
                List list = IteratorUtils.toList(iterator);
                Collections.sort(list);

               for(int i = list.size() - 1 ; i > 1 ; i --){

                   System.out.println(list.get(i));
                }
            }
        });
        jsc.close();

    }
}
